package com.flink.datastreamapi.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class GeneratorSource {


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //如果有n个并行度，最大值设置为a，将数值均分成n份，a/n,比如当最大为100 ，并行度为2时，每个并行度生成50个，其中一个是0-49 ，另一个是50-99
        env.setParallelism(1);//为了方便观察先设置并行度为1

        /*数据生成器Source 四个参数
          第一个 ，GeneratorFunction接口，需要实现，重写map方法，输入类型固定是Long
          第二个，long类型，自动生成的数字序列（从0自增）到最大值-1，达到这个值后就停止（如果想无界的效果可以设置Long.MAX_VALUE）
          第三个，限速策略，比如每秒生成几条数据
          第四个，返回类型
        */


        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource<>(new GeneratorFunction<Long, String>() {
            @Override
            public String map(Long value) throws Exception {
                return "Number:" + value;
            }
        }, 10, RateLimiterStrategy.perSecond(1), Types.STRING);

        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(),"data-genrator").print();

        env.execute();

    }
}
